package top.jolyoulu.core.rdd.serial

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * @Author: JolyouLu
 * @Date: 2024/5/12 17:06
 * @Description
 */
object Spark01_RDD_Serial {
  def main(args: Array[String]): Unit = {
    //准备环境 [*]:表示使用当前系统最大核
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    //替换默认序列化规则
    sparkConf.set("spark.serializer","com.esotericsoftware.kryo.KryoSerializable")
    //注册需要使用kryo序列化的自定义类
    sparkConf.registerKryoClasses(Array(classOf[Search]))

    val sc: SparkContext = new SparkContext(sparkConf)
    val rdd: RDD[String] = sc.makeRDD(Array("hello word", "hello spark", "hello"))
    //触发job的执行
    val search = new Search("h");

    search.getMatch1(rdd).collect().foreach(println)

    sc.stop()
  }

  class Search(query: String) extends Serializable {
    def isMatch(s: String): Boolean = {
      s.contains(query)
    }

    def getMatch1(rdd: RDD[String]): RDD[String] = {
      rdd.filter(isMatch)
    }

    def getMatch2(rdd: RDD[String]): RDD[String] = {
      rdd.filter(x => x.contains(query))
    }
  }
}

